package com.hhf.rrd.product;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

/**
 * 计算每一种商品（sku_id 是唯一标识）指标
 *      商品的出售个数、总销售额、平均销售额、最低价、最高价
 *
 * @author huanghaifeng15
 * @date 2022/2/15 14:06
 **/
public class ProductIndexQueryJob {

    public static void main(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inStreamingMode() // 声明为流任务
                //.inBatchMode() // 声明为批任务
                .build();

        TableEnvironment tEnv = TableEnvironment.create(settings);

        // 1. 创建一个数据源（输入）表，这里的数据源是 flink 自带的一个随机 mock 数据的数据源。
        String sourceSql = "CREATE TABLE source_table (\n"
                + "    sku_id STRING,\n"
                + "    price BIGINT\n"
                + ") WITH (\n"
                + "  'connector' = 'datagen',\n"
                + "  'rows-per-second' = '1',\n"
                + "  'fields.sku_id.length' = '1',\n"
                + "  'fields.price.min' = '1',\n"
                + "  'fields.price.max' = '1000000'\n"
                + ")";

        // 3. 执行一段 group by 的聚合 SQL 查询
        String selectWhereSql = "select sku_id,\n"
                + "       count(*) as count_result,\n"
                + "       sum(price) as sum_result,\n"
                + "       avg(price) as avg_result,\n"
                + "       min(price) as min_result,\n"
                + "       max(price) as max_result\n"
                + "from source_table\n"
                + "group by sku_id";

        tEnv.executeSql(sourceSql);
        tEnv.executeSql(selectWhereSql).print();
    }
}
